import pytest
import json
import os
from pathlib import (
    Path,
)
import re
import subprocess
import time
import zipfile

from eth_utils import (
    is_dict,
)
import pytest_asyncio

from web3._utils.contract_sources.contract_data.emitter_contract import (
    EMITTER_CONTRACT_DATA,
)
from web3._utils.contract_sources.contract_data.math_contract import (
    MATH_CONTRACT_DATA,
)
from web3._utils.contract_sources.contract_data.panic_errors_contract import (
    PANIC_ERRORS_CONTRACT_DATA,
)
from web3._utils.contract_sources.contract_data.revert_contract import (
    REVERT_CONTRACT_DATA,
)
from web3._utils.contract_sources.contract_data.storage_contract import (
    STORAGE_CONTRACT_DATA,
)

from .utils import (
    kill_proc_gracefully,
)

KEYFILE_PW = "web3py-test"
GETH_FIXTURE_ZIP = "geth-1.16.5-fixture.zip"


@pytest.fixture
def geth_binary():
    from geth.install import (
        get_executable_path,
        install_geth,
    )

    if "GETH_BINARY" in os.environ:
        return os.environ["GETH_BINARY"]
    elif "GETH_VERSION" in os.environ:
        geth_version = os.environ["GETH_VERSION"]
        _geth_binary = get_executable_path(geth_version)
        if not os.path.exists(_geth_binary):
            install_geth(geth_version)
        assert os.path.exists(_geth_binary)
        return _geth_binary
    else:
        return "geth"


def absolute_datadir(directory_name):
    return os.path.abspath(
        os.path.join(
            os.path.dirname(__file__),
            "..",
            directory_name,
        )
    )


@pytest.fixture
def get_geth_version(geth_binary):
    from geth import (
        get_geth_version,
    )

    geth_version = get_geth_version(geth_executable=os.path.expanduser(geth_binary))

    fixture_geth_version = GETH_FIXTURE_ZIP.split("-")[1]
    if fixture_geth_version not in str(geth_version):
        raise AssertionError(
            f"geth fixture version `{fixture_geth_version}` does not match geth "
            f"version for binary being used to run the test suite: `{geth_version}`. "
            "For CI runs, make sure to update the geth version in the CI config file."
        )

    return geth_version


@pytest.fixture
def base_geth_command_arguments(geth_binary, datadir):
    return (
        geth_binary,
        "--datadir",
        datadir,
        "--dev",
        "--dev.period",
        "3",
        "--password",
        os.path.join(datadir, "keystore", "pw.txt"),
        # in order to raise on underpriced transactions, ``txpool.nolocals`` is now
        # necessary: https://github.com/ethereum/go-ethereum/pull/31202
        "--txpool.nolocals",
    )


@pytest.fixture
def geth_zipfile_version(get_geth_version):
    # TODO: Remove support for 13 + 14 in next major version
    if get_geth_version.major == 1 and get_geth_version.minor in [13, 14, 15, 16]:
        return GETH_FIXTURE_ZIP
    raise AssertionError("Unsupported geth version")


@pytest.fixture
def datadir(tmpdir_factory, geth_zipfile_version):
    zipfile_path = absolute_datadir(geth_zipfile_version)
    base_dir = tmpdir_factory.mktemp("goethereum")
    tmp_datadir = os.path.join(str(base_dir), "datadir")
    with zipfile.ZipFile(zipfile_path, "r") as zip_ref:
        zip_ref.extractall(tmp_datadir)
    return tmp_datadir


@pytest.fixture
def geth_fixture_data(datadir):
    config_file_path = Path(datadir) / "config.json"
    return json.loads(config_file_path.read_text())


@pytest.fixture
def genesis_file(datadir):
    genesis_file_path = os.path.join(datadir, "genesis.json")
    return genesis_file_path


def wait_for_port(proc, timeout=10):
    start = time.time()
    wait_time = start + timeout

    while time.time() < wait_time:
        line = proc.stderr.readline()
        if not line:
            continue
        elif "geth.ipc" in line:
            # ipc path will be retrieved via ``get_ipc_path()`` fixture
            return None
        elif match := re.compile(r"127\.0\.0\.1:(\d+)").search(line):
            port = int(match.group(1))
            if port not in {0, 80}:
                # remove false positive matches
                return port

    raise TimeoutError(f"Did not find port in logs within {timeout} seconds")


@pytest.fixture
def start_geth_process_and_yield_port(
    geth_binary, datadir, genesis_file, geth_command_arguments
):
    init_cmd = (
        geth_binary,
        "--datadir",
        str(datadir),
        "init",
        str(genesis_file),
    )
    subprocess.check_output(init_cmd, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
    proc = subprocess.Popen(
        geth_command_arguments,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        bufsize=1,
    )

    try:
        port = wait_for_port(proc)
        yield port
    finally:
        # Ensure cleanup happens even if test fails
        kill_proc_gracefully(proc)
        output, errors = proc.communicate(timeout=5)
        print("Geth Process Exited:\n" f"stdout: {output}\n\n" f"stderr: {errors}\n\n")


@pytest.fixture
def math_contract_deploy_txn_hash(geth_fixture_data):
    return geth_fixture_data["math_deploy_txn_hash"]


@pytest.fixture
def math_contract(math_contract_factory, geth_fixture_data):
    return math_contract_factory(address=geth_fixture_data["math_address"])


@pytest.fixture
def math_contract_address(math_contract, address_conversion_func):
    return address_conversion_func(math_contract.address)


@pytest.fixture
def emitter_contract(emitter_contract_factory, geth_fixture_data):
    return emitter_contract_factory(address=geth_fixture_data["emitter_address"])


@pytest.fixture
def emitter_contract_address(emitter_contract, address_conversion_func):
    return address_conversion_func(emitter_contract.address)


@pytest.fixture
def keyfile_account_pkey(geth_fixture_data):
    return geth_fixture_data["keyfile_account_pkey"]


@pytest.fixture
def keyfile_account_address(geth_fixture_data):
    return geth_fixture_data["keyfile_account_address"]


@pytest.fixture
def keyfile_account_address_dual_type(keyfile_account_address, address_conversion_func):
    yield keyfile_account_address


@pytest.fixture
def empty_block(w3, geth_fixture_data):
    block = w3.eth.get_block(geth_fixture_data["empty_block_hash"])
    assert is_dict(block)
    return block


@pytest.fixture
def block_with_txn(w3, geth_fixture_data):
    block = w3.eth.get_block(geth_fixture_data["block_with_txn_hash"])
    assert is_dict(block)
    return block


@pytest.fixture
def mined_txn_hash(geth_fixture_data):
    return geth_fixture_data["mined_txn_hash"]


@pytest.fixture
def block_with_txn_with_log(w3, geth_fixture_data):
    block = w3.eth.get_block(geth_fixture_data["block_hash_with_log"])
    assert is_dict(block)
    return block


@pytest.fixture
def txn_hash_with_log(geth_fixture_data):
    return geth_fixture_data["txn_hash_with_log"]


@pytest.fixture
def block_hash_revert_no_msg(geth_fixture_data):
    return geth_fixture_data["block_hash_revert_no_msg"]


@pytest.fixture
def block_hash_revert_with_msg(geth_fixture_data):
    return geth_fixture_data["block_hash_revert_with_msg"]


@pytest.fixture
def revert_contract(revert_contract_factory, geth_fixture_data):
    return revert_contract_factory(address=geth_fixture_data["revert_address"])


@pytest.fixture
def offchain_lookup_contract(offchain_lookup_contract_factory, geth_fixture_data):
    return offchain_lookup_contract_factory(
        address=geth_fixture_data["offchain_lookup_address"]
    )


@pytest.fixture
def panic_errors_contract(
    w3,
    geth_fixture_data,
):
    contract_factory = w3.eth.contract(**PANIC_ERRORS_CONTRACT_DATA)
    return contract_factory(address=geth_fixture_data["panic_errors_contract_address"])


@pytest.fixture
def storage_contract(
    w3,
    geth_fixture_data,
):
    contract_factory = w3.eth.contract(**STORAGE_CONTRACT_DATA)
    return contract_factory(address=geth_fixture_data["storage_contract_address"])


# --- async --- #


@pytest_asyncio.fixture
async def async_keyfile_account_address(geth_fixture_data):
    return geth_fixture_data["keyfile_account_address"]


@pytest_asyncio.fixture
async def async_keyfile_account_address_dual_type(
    async_keyfile_account_address, address_conversion_func
):
    yield async_keyfile_account_address


@pytest.fixture
def async_offchain_lookup_contract(
    async_offchain_lookup_contract_factory, geth_fixture_data
):
    return async_offchain_lookup_contract_factory(
        address=geth_fixture_data["offchain_lookup_address"]
    )


@pytest.fixture
def async_panic_errors_contract(
    async_w3,
    geth_fixture_data,
):
    contract_factory = async_w3.eth.contract(**PANIC_ERRORS_CONTRACT_DATA)
    return contract_factory(address=geth_fixture_data["panic_errors_contract_address"])


@pytest.fixture
def async_emitter_contract(async_w3, geth_fixture_data):
    contract_factory = async_w3.eth.contract(**EMITTER_CONTRACT_DATA)
    return contract_factory(address=geth_fixture_data["emitter_address"])


@pytest.fixture
def async_emitter_contract_address(async_emitter_contract, address_conversion_func):
    return address_conversion_func(async_emitter_contract.address)


@pytest.fixture
def async_math_contract(async_w3, geth_fixture_data):
    contract_factory = async_w3.eth.contract(**MATH_CONTRACT_DATA)
    return contract_factory(address=geth_fixture_data["math_address"])


@pytest.fixture
def async_math_contract_address(async_math_contract, address_conversion_func):
    return address_conversion_func(async_math_contract.address)


@pytest.fixture
def async_revert_contract(async_w3, geth_fixture_data):
    contract_factory = async_w3.eth.contract(**REVERT_CONTRACT_DATA)
    return contract_factory(address=geth_fixture_data["revert_address"])


@pytest.fixture
def async_storage_contract(
    async_w3,
    geth_fixture_data,
):
    contract_factory = async_w3.eth.contract(**STORAGE_CONTRACT_DATA)
    return contract_factory(address=geth_fixture_data["storage_contract_address"])


@pytest_asyncio.fixture
async def async_empty_block(async_w3, geth_fixture_data):
    block = await async_w3.eth.get_block(geth_fixture_data["empty_block_hash"])
    assert is_dict(block)
    return block


@pytest_asyncio.fixture
async def async_block_with_txn(async_w3, geth_fixture_data):
    block = await async_w3.eth.get_block(geth_fixture_data["block_with_txn_hash"])
    assert is_dict(block)
    return block


@pytest_asyncio.fixture
async def async_block_with_txn_with_log(async_w3, geth_fixture_data):
    block = await async_w3.eth.get_block(geth_fixture_data["block_hash_with_log"])
    assert is_dict(block)
    return block
